package com.example.demo;

import jakarta.annotation.PostConstruct;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.List;

@Component
public class DeadLetterQueueConsumer {

    private static final Logger log = LoggerFactory.getLogger(DeadLetterQueueConsumer.class);

    private DefaultMQPushConsumer dlqConsumer;

    @PostConstruct
    public void init() throws MQClientException {
        dlqConsumer = new DefaultMQPushConsumer("dlq-consumer-group");
        dlqConsumer.setNamesrvAddr("192.168.10.32:9876");

        // 订阅死信队列
        dlqConsumer.subscribe("%DLQ%test-consumer-group", "*");
        dlqConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        dlqConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    String messageBody = new String(msg.getBody(), StandardCharsets.UTF_8);
                    log.error("死信队列消费: {}", messageBody);
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        dlqConsumer.start();
        log.info("RocketMQ 死信队列消费者已启动");
    }
}

